package com.github.yoojia.flow.zmq;

import com.github.yoojia.flow.Context;
import com.github.yoojia.flow.Event;
import com.github.yoojia.flow.VirtualDevice;
import com.github.yoojia.flow.util.ShutdownHook;
import com.parkingwang.lang.data.Config;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.zeromq.ZMQ;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static com.github.yoojia.flow.util.ConfigKit.checkRequired;

/**
 * 提供基于ZeroMQ通讯能力的虚拟设备实现类
 *
 * @author 陈哈哈 yoojiachen@gmail.com
 */
public class ZMQBridgeDevice extends VirtualDevice {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZMQBridgeDevice.class);

    private static final ZMQ.Context MQ_CONTEXT = ZMQ.context(2);

    static {
        ShutdownHook.add(MQ_CONTEXT::term);
    }

    public static final String FIELD_NAME_DRIVER_ADDRESS = "driver.address";
    public static final String FIELD_NAME_EVENT_ADDRESS = "events.bind";
    public static final String FIELD_NAME_EVENT_TOPIC = "events.topic";

    private final ZMQ.Socket mRequestSocket = MQ_CONTEXT.socket(ZMQ.REQ);
    private final ExecutorService mSubThread = Executors.newSingleThreadExecutor();
    private final ZMQ.Socket mSubSocket = MQ_CONTEXT.socket(ZMQ.SUB);

    @Override
    public void onInit(Context context, Config configs) throws Exception {
        super.onInit(context, configs);
        // 检查Config
        final String tag = ZMQBridgeDevice.class.getSimpleName();
        checkRequired(configs, FIELD_NAME_DRIVER_ADDRESS, tag);
        checkRequired(configs, FIELD_NAME_EVENT_ADDRESS, tag);
        checkRequired(configs, FIELD_NAME_EVENT_TOPIC, tag);
    }

    @Override
    public void onStart() {
        final Config config = getInitConfig();
        // 连接到远程设备进行通讯
        mRequestSocket.connect(config.getString(FIELD_NAME_DRIVER_ADDRESS));
        // 收到Sub通道发来的请求，转给总框架处理
        mSubSocket.bind(config.getString(FIELD_NAME_EVENT_ADDRESS));
        // 当Sub通道发送消息时，转给框架处理
        final String eventTopic = config.getString(FIELD_NAME_EVENT_TOPIC);
        mSubSocket.subscribe(eventTopic);

        mSubThread.submit(() -> {
            while (!Thread.currentThread().isInterrupted()) {
                // 接收Sub通道发来的消息请求
                final byte[] payload = mSubSocket.recv();
                final Event event = Event.createNoneDestAddress(
                        getContext().nextSequenceId(),
                        payload,
                        eventTopic,
                        getAddress(),
                        System.currentTimeMillis());
                LOGGER.debug("## SubSocket Received event: {}", event);
                getContext().publishEvent(event);
            }
        });
    }

    @Override
    public void onStop() {
        mRequestSocket.close();
        mSubSocket.close();
        mSubThread.shutdown();
    }

    @Override
    public Event handleEvent(Context context, Event request) {
        // 框架下发的请求消息，通过ZMQ转发给对应驱动设备来执行，并返回响应结果。
        // ZMQ发送内容： Payload字节内容
        // ZMQ返回内容： Payload字节内容
        mRequestSocket.send(request.payload);
        final byte[] response = mRequestSocket.recv();
        return createEvent(
                response,
                request.topic,
                getAddress(),
                request.sourceAddress);
    }
}
